package spark.sparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 从 socket 接收数据 进行词频统计
  */
object sparkStreaming01WordCount {
	def main(args: Array[String]): Unit = {
		val conf: SparkConf = new SparkConf().setAppName("wordcount").setMaster("local[*]")
		
		// 构建 StreamingContext
		val streamingContext = new StreamingContext(conf, Seconds(3))
		
		// 从指定的端口采集数据,
		val socketLineDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("nn1.hadoop", 9999)
		
		// 将采集的数据进行分解
		val wordDStream: DStream[String] = socketLineDStream.flatMap(_.split(" "))
		
		// 将结构转换便于统计
		val mapDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
		
		// reduceByKey 聚合
		val sumDStream: DStream[(String, Int)] = mapDStream.reduceByKey(_+_)
		
		// 打印结果，在Driver打印
		sumDStream.print()
		
		// 不能停止采集程序
		//streamingContext.stop()
		
		// 启动接收器
		streamingContext.start()
		// Driver等待接收器停止
		streamingContext.awaitTermination()
	}
}
